package com.clp.protocol.iec104.iec104.client;

import com.clp.protocol.core.connect.CouldNotAutoReconnectException;
import com.clp.protocol.core.exception.FailedToConnectException;
import com.clp.protocol.iec104.iec104.client.async.DefaultMasterPromise;
import com.clp.protocol.iec104.iec104.client.async.MasterFuture;
import com.clp.protocol.iec104.iec104.client.async.MasterFutureListener;
import com.clp.protocol.iec104.iec104.client.async.MasterPromise;
import com.clp.protocol.iec104.iec104.client.event.MasterConnectEvent;
import com.clp.protocol.iec104.iec104.client.pipeline.Iec104ChannelInitializer;
import com.clp.protocol.iec104.iec104.client.pipeline.handler.MasterHandlersSupplier;
import com.clp.util.event.EventPublisher;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Nullable;

@Slf4j
public class InMasterConnector {
    private final InMaster inMaster;
    private final Iec104ChannelInitializer initializer;
    @Getter
    private final EventPublisher<Master> eventPublisher;
    private volatile boolean canAutoReconnect = false;

    private volatile MasterFuture<Void> future;
    private volatile ConnectingState currState = ConnectingState.NONE;

    public InMasterConnector(InMaster inMaster, MasterHandlersSupplier supplier) {
        this.inMaster = inMaster;
        this.initializer = new Iec104ChannelInitializer(inMaster, supplier);
        this.eventPublisher = new EventPublisher<>(inMaster.getAsyncSupport().scheduledExecutorService());
    }

    /**
     * 如果是无状态，那么返回null；如果是当前状态，那么返回该状态的future
     *
     * @param state
     * @return
     */
    @Nullable
    private synchronized MasterFuture<Void> checkStateAndGetFuture(ConnectingState state) {
        if (currState == ConnectingState.NONE) {
            return null;
        }
        if (currState == state) {
            return future;
        }
        log.warn("当前状态：" + currState + ", 需要状态：" + state);
        throw new IllegalStateException("当前状态：" + currState + ", 需要状态：" + state);
    }

    private synchronized void setStateAndFuture(ConnectingState state, MasterFuture<Void> future) {
        this.currState = state;
        this.future = future;
    }

    /**
     * 发布连接事件
     *
     * @param cause 如果是失败事件，说明原因
     * @param type  连接事件的类型
     */
    private void publishConnectEvent(MasterConnectEvent.Type type, @Nullable Throwable cause) {
        eventPublisher.publishEvent(new MasterConnectEvent(inMaster, type, cause));
    }

    private void publishConnectEvent(MasterConnectEvent.Type type) {
        eventPublisher.publishEvent(new MasterConnectEvent(inMaster, type, null));
    }

    public MasterFuture<Void> connect() {
        if (inMaster.isConnected()) {
            throw new IllegalStateException("当前连接有效，无法再次进行连接！");
        }
        MasterFuture<Void> future = checkStateAndGetFuture(ConnectingState.CONNECTING);
        if (future != null) {
            return future;
        }
        synchronized (this) {
            MasterPromise<Void> promise = new DefaultMasterPromise<>(inMaster);
            publishConnectEvent(MasterConnectEvent.Type.CONNECT_BEFORE);
            doBeforeConnect();
            connect0().addListener(new MasterFutureListener<Channel>() {
                @Override
                public void operationComplete(MasterFuture<Channel> future) {
                    if (!future.isSuccess()) {
                        promise.setFailure(future.cause());
                        return;
                    }
                    doAfterConnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                        @Override
                        public void operationComplete(Future<? super Void> future) throws Exception {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            promise.setSuccess();
                        }
                    });
                }
            });
            setStateAndFuture(ConnectingState.CONNECTING, promise);
            return promise.addListener(new MasterFutureListener<Void>() {
                @Override
                public void operationComplete(MasterFuture<Void> future) {
                    synchronized (InMasterConnector.this) {
                        if (future.isSuccess()) {
                            publishConnectEvent(MasterConnectEvent.Type.CONNECT_SUCCESS);
                        } else {
                            publishConnectEvent(MasterConnectEvent.Type.CONNECT_FAILED, future.cause());
                        }
                        setStateAndFuture(ConnectingState.NONE, null);
                    }
                }
            });
        }
    }

    /**
     * 连接：会先进行一次连接，如果连接成功，返回成功结果；如果没有成功，则会一直重连，直到超过最大重连次数
     *
     * @return
     */
    private MasterFuture<Channel> connect0() {
        MasterPromise<Channel> promise = new DefaultMasterPromise<>(inMaster);
        inMaster.getNettyClient().connect(initializer, inMaster.remoteHost(), inMaster.remotePort()).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    promise.setRes(future.channel());
                    promise.setSuccess();
                    log.info("连接{}:{}成功", inMaster.remoteHost(), inMaster.remotePort());
                } else {
                    future.channel().close();
                    log.warn("连接{}:{}失败, 原因：{}，尝试重新连接", inMaster.remoteHost(), inMaster.remotePort(), future.cause().getMessage());
                    connect0Loop(1).addListener(new MasterFutureListener<Channel>() {
                        @Override
                        public void operationComplete(MasterFuture<Channel> future) {
                            if (future.isSuccess()) {
                                promise.setRes(future.getRes());
                                promise.setSuccess();
                                log.info("重连 {}:{} 成功！", inMaster.remoteHost(), inMaster.remotePort());
                            } else {
                                promise.setFailure(future.cause());
                                log.error("连接 {}:{} 失败，原因：{}", inMaster.remoteHost(), inMaster.remotePort(), future.cause().getMessage());
                            }
                        }
                    });
                }
            }
        });
        return promise;
    }

    private MasterFuture<Channel> connect0Loop(int reconnCount) {
        MasterPromise<Channel> promise = new DefaultMasterPromise<>(inMaster);
        try {
            if (inMaster.maxReconnectCount() >= 0 && reconnCount >= inMaster.maxReconnectCount()) {
                promise.setFailure(new FailedToConnectException("超过最大重连次数，连接失败！"));
                return promise;
            }

            log.info("进行对 {}:{} 的第{}次重连", inMaster.remoteHost(), inMaster.remotePort(), reconnCount);
            Thread.sleep(reconnCount * 1000L);

            inMaster.getNettyClient().connect(initializer, inMaster.remoteHost(), inMaster.remotePort()).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        log.info("对 {}:{} 的第{}次重连成功", inMaster.remoteHost(), inMaster.remotePort(), reconnCount);
                        promise.setRes(future.channel());
                        promise.setSuccess();
                    } else {
                        future.channel().close();
                        connect0Loop(reconnCount + 1).addListener(new MasterFutureListener<Channel>() {
                            @Override
                            public void operationComplete(MasterFuture<Channel> future) {
                                if (future.isSuccess()) {
                                    promise.setRes(future.getRes());
                                    promise.setSuccess();
                                } else {
                                    promise.setFailure(future.cause());
                                }
                            }
                        });
                    }
                }
            });
            return promise;
        } catch (InterruptedException e) {
            promise.setFailure(e);
            return promise;
        }
    }

    private synchronized void doBeforeConnect() {
        inMaster.resetSeqs();
    }

    private synchronized Future<Void> doAfterConnectSuccess(Channel channel) {
        Promise<Void> promise = inMaster.getAsyncSupport().createPromise(Void.class);
        inMaster.resetChannel(channel);
        canAutoReconnect = true;
        inMaster.stateInfo().activate().addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess()) {
                    promise.setSuccess(null);
                    return;
                }
                if (!inMaster.isConnected()) {
                    // 尝试重新链接
                    connect0().addListener(new MasterFutureListener<Channel>() {
                        @Override
                        public void operationComplete(MasterFuture<Channel> future) {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            doAfterConnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                                @Override
                                public void operationComplete(Future<? super Void> future) throws Exception {
                                    if (future.isSuccess()) {
                                        promise.setSuccess(null);
                                        return;
                                    }
                                    promise.setFailure(future.cause());
                                }
                            });
                        }
                    });
                } else {
                    doAfterConnectSuccess(channel).addListener(new GenericFutureListener<Future<? super Void>>() {
                        @Override
                        public void operationComplete(Future<? super Void> future) throws Exception {
                            if (future.isSuccess()) {
                                promise.setSuccess(null);
                                return;
                            }
                            promise.setFailure(future.cause());
                        }
                    });
                }
            }
        });
        return promise;
    }

    public void tryAutoReconnect() {
        try {
            autoReconnect();
        } catch (CouldNotAutoReconnectException ex) {
            log.info("取消自动重连，原因：" + ex.getMessage());
        }
    }

    /**
     * @return
     * @throws CouldNotAutoReconnectException
     */
    private synchronized MasterFuture<Void> autoReconnect() throws CouldNotAutoReconnectException {
        if (!canAutoReconnect) {
            log.warn("当前无法自动重连");
            throw new CouldNotAutoReconnectException("The channel could not auto reconnect");
        }
        try {
            MasterFuture<Void> future = checkStateAndGetFuture(ConnectingState.AUTO_RECONNECTING);
            if (future != null) {
                return future;
            }
        } catch (IllegalStateException e) {
            throw new CouldNotAutoReconnectException(e);
        }
        MasterPromise<Void> promise = new DefaultMasterPromise<>(inMaster);
        publishConnectEvent(MasterConnectEvent.Type.AUTO_RECONNECT_BEFORE);
        doBeforeAutoReconnect().addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (!future.isSuccess()) {
                    promise.setFailure(future.cause());
                    return;
                }
                if (inMaster.isConnected()) {
                    disconnect0().addListener(new GenericFutureListener<Future<? super Void>>() {
                        @Override
                        public void operationComplete(Future<? super Void> future) throws Exception {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            connect0().addListener(new MasterFutureListener<Channel>() {
                                @Override
                                public void operationComplete(MasterFuture<Channel> future) {
                                    if (!future.isSuccess()) {
                                        promise.setFailure(future.cause());
                                        return;
                                    }
                                    doAfterAutoReconnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                                        @Override
                                        public void operationComplete(Future<? super Void> future) throws Exception {
                                            if (!future.isSuccess()) {
                                                // 如果重连后处理失败，那么
                                                promise.setFailure(future.cause());
                                                return;
                                            }
                                            promise.setSuccess();
                                        }
                                    });
                                }
                            });
                        }
                    });
                } else {
                    connect0().addListener(new MasterFutureListener<Channel>() {
                        @Override
                        public void operationComplete(MasterFuture<Channel> future) {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            doAfterAutoReconnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                                @Override
                                public void operationComplete(Future<? super Void> future) throws Exception {
                                    if (!future.isSuccess()) {
                                        promise.setFailure(future.cause());
                                        return;
                                    }
                                    promise.setSuccess();
                                }
                            });
                        }
                    });
                }
            }
        });
        setStateAndFuture(ConnectingState.AUTO_RECONNECTING, promise);
        return promise.addListener(new MasterFutureListener<Void>() {
            @Override
            public void operationComplete(MasterFuture<Void> future) {
                synchronized (InMasterConnector.this) {
                    if (future.isSuccess()) {
                        publishConnectEvent(MasterConnectEvent.Type.AUTO_RECONNECT_SUCCESS);
                    } else {
                        publishConnectEvent(MasterConnectEvent.Type.AUTO_RECONNECT_FAILED);
                    }
                    setStateAndFuture(ConnectingState.NONE, null);
                    if (!future.isSuccess() && !inMaster.isConnected()) {
                        log.info("尝试自动重连失败，原因：{}，即将断开这个连接", future.cause().getMessage());
                        autoDisconnect();
                    }
                }
            }
        });
    }

    private Future<Void> doBeforeAutoReconnect() {
        return doBeforeReconnect();
    }

    private Future<Void> doAfterAutoReconnectSuccess(Channel channel) {
        return doAfterReconnectSuccess(channel);
    }

    /**
     * 客户端主动重连
     *
     * @return
     */
    public synchronized MasterFuture<Void> reconnect() {
        MasterFuture<Void> future = checkStateAndGetFuture(ConnectingState.RECONNECTING);
        if (future != null) {
            return future;
        }
        synchronized (this) {
            MasterPromise<Void> promise = new DefaultMasterPromise<>(inMaster);
            publishConnectEvent(MasterConnectEvent.Type.RECONNECT_BEFORE);
            doBeforeReconnect().addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if (!future.isSuccess()) {
                        promise.setFailure(future.cause());
                        return;
                    }
                    if (inMaster.isConnected()) {
                        disconnect0().addListener(new GenericFutureListener<Future<? super Void>>() {
                            @Override
                            public void operationComplete(Future<? super Void> future) throws Exception {
                                if (!future.isSuccess()) {
                                    promise.setFailure(future.cause());
                                    return;
                                }
                                connect0().addListener(new MasterFutureListener<Channel>() {
                                    @Override
                                    public void operationComplete(MasterFuture<Channel> future) {
                                        if (!future.isSuccess()) {
                                            promise.setFailure(future.cause());
                                            return;
                                        }
                                        doAfterReconnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                                            @Override
                                            public void operationComplete(Future<? super Void> future) throws Exception {
                                                if (!future.isSuccess()) {
                                                    promise.setFailure(future.cause());
                                                    return;
                                                }
                                                promise.setSuccess();
                                            }
                                        });
                                    }
                                });
                            }
                        });
                    } else {
                        connect0().addListener(new MasterFutureListener<Channel>() {
                            @Override
                            public void operationComplete(MasterFuture<Channel> future) {
                                if (!future.isSuccess()) {
                                    promise.setFailure(future.cause());
                                    return;
                                }
                                doAfterReconnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                                    @Override
                                    public void operationComplete(Future<? super Void> future) throws Exception {
                                        if (!future.isSuccess()) {
                                            promise.setFailure(future.cause());
                                            return;
                                        }
                                        promise.setSuccess();
                                    }
                                });
                            }
                        });
                    }
                }
            });
            setStateAndFuture(ConnectingState.RECONNECTING, promise);
            return promise.addListener(new MasterFutureListener<Void>() {
                @Override
                public void operationComplete(MasterFuture<Void> future) {
                    synchronized (InMasterConnector.this) {
                        if (future.isSuccess()) {
                            publishConnectEvent(MasterConnectEvent.Type.RECONNECT_SUCCESS);
                        } else {
                            publishConnectEvent(MasterConnectEvent.Type.RECONNECT_FAILED);
                        }
                        setStateAndFuture(ConnectingState.NONE, null);
                    }
                }
            });
        }
    }

    private Future<Void> doBeforeReconnect() {
        return doBeforeDisconnect();
    }

    private synchronized Future<Void> doAfterReconnectSuccess(Channel channel) {
        Promise<Void> promise = inMaster.getAsyncSupport().createPromise(Void.class);
        inMaster.resetChannel(channel);
        canAutoReconnect = true;
        Future<Void> future = inMaster.stateInfo().reactivate();
        future.addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess()) {
                    promise.setSuccess(null);
                    return;
                }
                if (!inMaster.isConnected()) {
                    // 此处再次尝试重新重连
                    connect0().addListener(new MasterFutureListener<Channel>() {
                        @Override
                        public void operationComplete(MasterFuture<Channel> future) {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            doAfterReconnectSuccess(future.getRes()).addListener(new GenericFutureListener<Future<? super Void>>() {
                                @Override
                                public void operationComplete(Future<? super Void> future) throws Exception {
                                    if (future.isSuccess()) {
                                        promise.setSuccess(null);
                                        return;
                                    }
                                    promise.setFailure(future.cause());
                                }
                            });
                        }
                    });
                } else {
                    doAfterReconnectSuccess(channel).addListener(new GenericFutureListener<Future<? super Void>>() {
                        @Override
                        public void operationComplete(Future<? super Void> future) throws Exception {
                            if (future.isSuccess()) {
                                promise.setSuccess(null);
                                return;
                            }
                            promise.setFailure(future.cause());
                        }
                    });
                }
            }
        });
        return promise;
    }

    private synchronized MasterFuture<Void> autoDisconnect() {
        MasterFuture<Void> future = checkStateAndGetFuture(ConnectingState.AUTO_DISCONNECTING);
        if (future != null) {
            return future;
        }
        synchronized (this) {
            DefaultMasterPromise<Void> promise = new DefaultMasterPromise<>(inMaster);
            publishConnectEvent(MasterConnectEvent.Type.AUTO_DISCONNECT_BEFORE);
            doBeforeAutoDisconnect().addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if (!future.isSuccess()) {
                        promise.setFailure(future.cause());
                        return;
                    }
                    if (!inMaster.isConnected()) {
                        log.info("连接 {}:{} 已断开！", inMaster.remoteHost(), inMaster.remotePort());
                        promise.setSuccess();
                    }
                    disconnect0().addListener(new GenericFutureListener<Future<? super Void>>() {
                        @Override
                        public void operationComplete(Future<? super Void> future) throws Exception {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            promise.setSuccess();
                        }
                    });
                }
            });
            setStateAndFuture(ConnectingState.AUTO_DISCONNECTING, promise);
            return promise.addListener(new MasterFutureListener<Void>() {
                @Override
                public void operationComplete(MasterFuture<Void> future) {
                    synchronized (InMasterConnector.this) {
                        if (future.isSuccess()) {
                            publishConnectEvent(MasterConnectEvent.Type.AUTO_DISCONNECT_SUCCESS);
                        } else {
                            publishConnectEvent(MasterConnectEvent.Type.AUTO_DISCONNECT_FAILED, future.cause());
                        }
                        setStateAndFuture(ConnectingState.NONE, null);
                    }
                }
            });
        }
    }

    private Future<Void> doBeforeAutoDisconnect() {
        return doBeforeDisconnect();
    }

    public synchronized MasterFuture<Void> disconnect() {
        MasterFuture<Void> future = checkStateAndGetFuture(ConnectingState.DISCONNECTING);
        if (future != null) {
            return future;
        }
        synchronized (this) {
            DefaultMasterPromise<Void> promise = new DefaultMasterPromise<>(inMaster);
            publishConnectEvent(MasterConnectEvent.Type.DISCONNECT_BEFORE);
            if (!inMaster.isConnected()) {
                log.info("连接 {}:{} 已断开！", inMaster.remoteHost(), inMaster.remotePort());
                promise.setSuccess();
                publishConnectEvent(MasterConnectEvent.Type.DISCONNECT_SUCCESS);
                return promise;
            }
            doBeforeDisconnect().addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if (!future.isSuccess()) {
                        promise.setFailure(future.cause());
                        return;
                    }
                    disconnect0().addListener(new GenericFutureListener<Future<? super Void>>() {
                        @Override
                        public void operationComplete(Future<? super Void> future) throws Exception {
                            if (!future.isSuccess()) {
                                promise.setFailure(future.cause());
                                return;
                            }
                            promise.setSuccess();
                        }
                    });
                }
            });
            setStateAndFuture(ConnectingState.DISCONNECTING, promise);
            return promise.addListener(new MasterFutureListener<Void>() {
                @Override
                public void operationComplete(MasterFuture<Void> future) {
                    synchronized (InMasterConnector.this) {
                        if (future.isSuccess()) {
                            publishConnectEvent(MasterConnectEvent.Type.DISCONNECT_SUCCESS);
                        } else {
                            publishConnectEvent(MasterConnectEvent.Type.DISCONNECT_FAILED, future.cause());
                        }
                        setStateAndFuture(ConnectingState.NONE, null);
                    }
                }
            });
        }
    }

    private ChannelFuture disconnect0() {
        return inMaster.channel().close().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("断开连接 {}:{} 成功！", inMaster.remoteHost(), inMaster.remotePort());
                } else {
                    log.error("断开连接 {}:{} 失败！", inMaster.remoteHost(), inMaster.remotePort());
                }
            }
        });
    }

    private synchronized Future<Void> doBeforeDisconnect() {
        inMaster.resetSeqs();
        canAutoReconnect = false;
        return inMaster.stateInfo().inactivate();
    }

    private enum ConnectingState {
        NONE,
        CONNECTING,
        RECONNECTING,
        AUTO_RECONNECTING,
        DISCONNECTING,
        AUTO_DISCONNECTING
    }

}
